#pragma once
#include <iostream>
#include <string>
#include <list>
#include "../mqcommon/Helper.hpp"
#include "../mqcommon/Log.hpp"
#include "../mqcommon/msg.pb.h"
using namespace std;
namespace mq
{
#define DATAFILE_SUBFIX ".mqd"
#define TMPFILE_SUBFIX ".mqd.tmp"
    class MessageMapper
    {
        using MessagePtr = shared_ptr<mq::MQMessage>;

    public:
        MessageMapper(string &basedir, const string &qname)
            : _qname(qname)
        {
            if (basedir.back() != '/')
                basedir += '/';
            _datafile = basedir + qname + DATAFILE_SUBFIX;
            _tmpfile = basedir + qname + TMPFILE_SUBFIX;
            mq::FileHelper helper(basedir);
            if (helper.exists() == false)
                mq::FileHelper::createDirectory(basedir);
            assert(createMsgFile());
        }
        // 创建持久化存储文件
        bool createMsgFile()
        {
            if (mq::FileHelper(_datafile).exists() == false)
            {
                if (mq::FileHelper::createFile(_datafile) == false)
                {
                    DLOG("创建文件失败");
                    return false;
                }
            }
            return true;
        }
        // 删除持久化存储文件
        void removeMsgFile()
        {
            mq::FileHelper::removeFile(_datafile);
            mq::FileHelper::removeFile(_tmpfile);
        }
        // 向持久化文件中插入消息
        bool insert(MessagePtr &msg)
        {
            return insert(_datafile, msg);
        }
        // 从持久化文件中删除消息
        bool remove(MessagePtr &msg)
        {
            msg->mutable_payload()->set_valid("0");
            string msg_str = msg->payload().SerializeAsString();
            mq::FileHelper helper(_datafile);
            size_t offset = msg->offset();
            size_t msize = msg_str.size();
            if (msize != msg->length())
            {
                DLOG("写入数据长度不一致");
                return false;
            }
            bool ret = helper.write(msg_str.c_str(), offset, msize);
            if (ret == false)
            {
                DLOG("文件写入失败");
                return false;
            }
            return true;
        }
        // 持久化文件垃圾回收
        list<MessagePtr> gc()
        {
            list<MessagePtr> result;
            bool ret = load(result);
            // DLOG("加载到消息个数: %ld", result.size());
            if (ret == false)
            {
                DLOG("数据加载失败");
                return result;
            }
            mq::FileHelper::createFile(_tmpfile);
            for (auto &mp : result)
            {
                if (insert(_tmpfile, mp) == false)
                {
                    DLOG("临时文件写入失败");
                    return result;
                }
                // DLOG("写入消息:%s", mp->payload().body().c_str());
            }

            if (mq::FileHelper::removeFile(_datafile) == false)
            {
                DLOG("删除源文件失败");
                return result;
            };
            if (mq::FileHelper(_tmpfile).rename(_datafile) == false)
            {
                DLOG("修改临时文件名称失败");
                return result;
            }
            return result;
        }

    private:
        bool load(std::list<MessagePtr> &result)
        {
            mq::FileHelper helper(_datafile);
            size_t offset = 0, msg_size = 0;
            size_t fsize = helper.size();
            while (offset < fsize)
            {
                if (helper.read((char *)&msg_size, offset, sizeof(size_t)) == false)
                {
                    DLOG("读取数据长度失败");
                    return false;
                }
                offset += sizeof(size_t);
                string msg_body(msg_size, '\0');
                if (helper.read(&msg_body[0], offset, msg_size) == false)
                {
                    DLOG("读取数据内容失败");
                    return false;
                }
                offset += msg_size;
                MessagePtr mp = make_shared<mq::MQMessage>();
                mp->mutable_payload()->ParseFromString(msg_body);
                if (mp->payload().valid() == "0")
                    continue;
                result.push_back(mp);
            }
            return true;
        }
        bool insert(const std::string &filename, MessagePtr &msg)
        {
            mq::FileHelper helper(filename);
            string msg_str = msg->payload().SerializeAsString();
            size_t fsize = helper.size();
            size_t msize = msg_str.size();
            bool ret = helper.write((char *)(&msize), fsize, sizeof(size_t));
            if (ret == false)
            {
                DLOG("文件写入失败");
                return false;
            }
            ret = helper.write(msg_str.c_str(), fsize + sizeof(size_t), msize);
            if (ret == false)
            {
                DLOG("文件写入失败");
                return false;
            }
            // msg 中的实际存储信息
            msg->set_offset(fsize + sizeof(size_t));
            msg->set_length(msize);
            // DLOG("向文件: %s 写入成功: %s", filename.c_str(), msg->payload().body().c_str());
            return true;
        }

    private:
        string _qname;
        string _datafile;
        string _tmpfile;
    };
    class QueueMessage
    {
    public:
        using MessagePtr = shared_ptr<mq::MQMessage>;
        using QueueMessagePtr = shared_ptr<QueueMessage>;
        QueueMessage(string &basedir, const std::string &qname)
            : _qname(qname),
              _valid_count(0), _total_count(0), _mapper(basedir, qname)
        {
        }
        // 从持久化文件恢复
        bool recovery()
        {
            unique_lock<mutex> lock(_mutex);
            _msgs = _mapper.gc();
            // DLOG("恢复队列消息个数: %ld", _msgs.size());
            for (auto &msg : _msgs)
            {
                _durable_msgs.insert(make_pair(msg->payload().properties().id(), msg));
            }
            _valid_count = _total_count = _msgs.size();
            return true;
        }
        // 向队列中插入一条消息
        bool insert(const BasicProperties *bp, const std::string &body, bool queue_is_durable)
        {
            MessagePtr mp = make_shared<MQMessage>();
            mp->mutable_payload()->set_body(body);
            if (bp)
            {
                DeliveryMode mode = queue_is_durable ? bp->delivery_mode() : DeliveryMode::UNDURABLE;
                mp->mutable_payload()->mutable_properties()->set_id(bp->id());
                mp->mutable_payload()->mutable_properties()->set_routing_key(bp->routing_key());
                mp->mutable_payload()->mutable_properties()->set_delivery_mode(mode);
            }
            else
            {
                DeliveryMode mode = queue_is_durable ? DeliveryMode::DURABLE : DeliveryMode::UNDURABLE;
                mp->mutable_payload()->mutable_properties()->set_id(UuidHelper::uuid());
                mp->mutable_payload()->mutable_properties()->set_routing_key("");
                mp->mutable_payload()->mutable_properties()->set_delivery_mode(mode);
            }
            unique_lock<mutex> lock(_mutex);
            if (mp->payload().properties().delivery_mode() == DeliveryMode::DURABLE)
            {
                mp->mutable_payload()->set_valid("1");
                if (_mapper.insert(mp) == false)
                {
                    DLOG("消息持久化失败");
                    return false;
                }
                _durable_msgs.insert(make_pair(mp->payload().properties().id(), mp));
                _valid_count += 1;
                _total_count += 1;
            }
            _msgs.push_back(mp);
            return true;
        }
        // 从队列中获取一个消息
        MessagePtr front()
        {
            unique_lock<mutex> lock(_mutex);
            if (_msgs.size() == 0)
                return MessagePtr();
            MessagePtr mp = _msgs.front();
            _msgs.pop_front();
            _waitack_msgs.insert(make_pair(mp->payload().properties().id(), mp));
            return mp;
        }
        // 向队列中消息进行应答
        bool remove(const std::string &msg_id)
        {
            unique_lock<mutex> lock(_mutex);
            auto it = _waitack_msgs.find(msg_id);
            if (it == _waitack_msgs.end())
            {
                DLOG("无法找到删除的消息：%s", msg_id.c_str());
                return true;
            }
            if (it->second->payload().properties().delivery_mode() == DeliveryMode::DURABLE)
            {
                _durable_msgs.erase(msg_id);
                _mapper.remove(it->second);
                _valid_count -= 1;
                gc();
            }
            _waitack_msgs.erase(it);
            return true;
        }
        // 获取可获取消息数量
        size_t getable_count()
        {
            unique_lock<mutex> lock(_mutex);
            return _msgs.size();
        }
        // 获取总持久化消息数量
        size_t total_count()
        {
            unique_lock<mutex> lock(_mutex);
            return _total_count;
        }
        // 获取有效持久化消息数量
        size_t durable_count()
        {
            unique_lock<mutex> lock(_mutex);
            return _durable_msgs.size();
        }
        // 获取等待确认消息数量
        size_t waitack_count()
        {
            unique_lock<mutex> lock(_mutex);
            return _waitack_msgs.size();
        }
        // 清空队列所有消息
        void clear()
        {
            unique_lock<mutex> lock(_mutex);
            _mapper.removeMsgFile();
            _msgs.clear();
            _durable_msgs.clear();
            _waitack_msgs.clear();
            _valid_count = _total_count = 0;
        }

    private:
        bool GCCheck()
        {
            if (_total_count > 2000 && _valid_count * 10 / _total_count < 5)
                return true;
            return false;
        }
        void gc()
        {
            if (GCCheck() == false)
                return;
            list<MessagePtr> msgs = _mapper.gc();
            for (auto &msg : msgs)
            {
                auto it = _durable_msgs.find(msg->payload().properties().id());
                if (it == _durable_msgs.end())
                {
                    DLOG("出现未知序列化消息:%s", msg->payload().properties().id().c_str());
                    _durable_msgs.insert(make_pair(msg->payload().properties().id(), msg));
                    continue;
                }
                it->second->set_offset(msg->offset());
                it->second->set_length(msg->length());
            }
            _valid_count = _total_count = msgs.size();
        }

    private:
        std::mutex _mutex;
        std::string _qname;
        size_t _valid_count; // 有效持久化消息数量
        size_t _total_count; // 总的持久化消息数量
        MessageMapper _mapper;
        list<MessagePtr> _msgs;                          // 待推送消息
        unordered_map<string, MessagePtr> _durable_msgs; // 持久化消息 hash
        unordered_map<string, MessagePtr> _waitack_msgs; // 待确认消息 has
    };
    class MessageManager
    {
    public:
        using MessageManagerPtr = shared_ptr<MessageManager>;
        using MessagePtr = shared_ptr<mq::MQMessage>;
        MessageManager(const std::string &basedir)
            : _basedir(basedir)
        {
        }
        // 初始化（新建）指定消息队列
        void initQueueMessage(const std::string &qname)
        {
            QueueMessage::QueueMessagePtr qmp;
            {
                unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it != _queue_msgs.end())
                    return;
                qmp = make_shared<QueueMessage>(_basedir, qname);
                _queue_msgs.insert(make_pair(qname, qmp));
            }
            if (qmp->recovery() == false)
            {
                DLOG("队列历史数据恢复失败:%s", qname.c_str());
            }
            // DLOG("队列历史数据恢复成功:%s", qname.c_str());
        }
        // 销毁指定消息队列
        void destroyQueueMessage(const std::string &qname)
        {
            QueueMessage::QueueMessagePtr qmp;
            {
                unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    return;
                }
                qmp = it->second;
            }
            qmp->clear();
        }
        // 向指定队列中插入消息
        bool insert(const std::string &qname, BasicProperties *bp, const std::string &body, bool queue_is_durable)
        {
            QueueMessage::QueueMessagePtr qmp;
            {
                unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("不存在目标队列: %s", qname.c_str());
                    return false;
                }
                qmp = it->second;
            }
            if (qmp->insert(bp, body, queue_is_durable) == false)
            {
                DLOG("向目标队列插入消息失败: %s", qname.c_str());
                return false;
            }
            // DLOG("向目标队列插入消息成功: %s->%s", qname.c_str(), body.c_str());
            return true;
        }
        // 从指定队列中获取消息
        MessagePtr front(const std::string &qname)
        {
            QueueMessage::QueueMessagePtr qmp;
            {
                unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("不存在目标队列: %s", qname.c_str());
                    return MessagePtr();
                }
                qmp = it->second;
            }
            return qmp->front();
        }
        // 向指定队列中确认消息
        bool ack(const std::string &qname, const std::string &msg_id)
        {
            QueueMessage::QueueMessagePtr qmp;
            {
                unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("不存在目标队列: %s", qname.c_str());
                    return false;
                }
                qmp = it->second;
            }
            if (qmp->remove(msg_id) == false)
            {
                DLOG("目标队列:%s 不存在目标消息: %s", qname.c_str(), msg_id.c_str());
                return false;
            }
            return true;
        }
        // 获取可获取消息数量
        size_t getable_count(const std::string &qname)
        {
            QueueMessage::QueueMessagePtr qmp;
            {
                unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("不存在目标队列: %s", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->getable_count();
        }
        // 获取可持久化总消息
        size_t total_count(const std::string &qname)
        {
            QueueMessage::QueueMessagePtr qmp;
            {
                unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("不存在目标队列: %s", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->total_count();
        }
        // 获取有效持久化消息数量
        size_t durable_count(const std::string &qname)
        {
            QueueMessage::QueueMessagePtr qmp;
            {
                unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("不存在目标队列: %s", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->durable_count();
        }
        // 获取等待确认消息数量
        size_t waitack_count(const std::string &qname)
        {
            QueueMessage::QueueMessagePtr qmp;
            {
                unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("不存在目标队列: %s", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->waitack_count();
        }
        // 清空所有队列的消息
        void clear()
        {
            unique_lock<std::mutex> lock(_mutex);
            for (auto &it : _queue_msgs)
                it.second->clear();
            _queue_msgs.clear();
        }

    private:
        mutex _mutex;
        string _basedir;
        unordered_map<string, QueueMessage::QueueMessagePtr> _queue_msgs; // 队列管理
    };
}